package com.group2.edu.realtime.dwd;/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 15:50
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.constant.Constant;

import com.group2.edu.realtime.common.util.DateFormatUtil;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 15:50
 * 需要启动进程：dwdbaselog，DwdTrafficUniqueVisitorDetail
 */
public class DwdTrafficUniqueVisitorDetail extends BaseApp {

    public static void main(String[] args) {
        new DwdTrafficUniqueVisitorDetail().start(10012,4,
                "DwdTrafficUniqueVisitorDetail",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> mainSource) {

        SingleOutputStreamOperator<JSONObject> jsonObjDs = mainSource.map(
                JSON::parseObject
        );
        //过滤last_page_id 不为null的数据
        SingleOutputStreamOperator<JSONObject> filterDs = jsonObjDs.filter(
                jsonObj -> jsonObj.getJSONObject("page").getString("last_page_id") == null
        );
        //按照mid分组
        KeyedStream<JSONObject, String> keyByDs = filterDs.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        //flink状态编程过滤独立访客记录
        SingleOutputStreamOperator<JSONObject> filterDs2 = keyByDs.filter(
                new RichFilterFunction<JSONObject>() {
                    private ValueState<String> lastVisitDt;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateProperties = new ValueStateDescriptor<String>("lastVisitDt", String.class);
                        stateProperties.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDt = getRuntimeContext().getState(stateProperties);


                    }

                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        String lastDt = lastVisitDt.value();
                        String visitDt = DateFormatUtil.tsToDate(jsonObj.getLong("ts"));
                        if (lastDt == null || (DateFormatUtil.dateToTs(lastDt) < DateFormatUtil.dateToTs(visitDt))) {
                            lastVisitDt.update(visitDt);
                            return true;
                        }
                        return false;

                    }
                }
        );
        //将独立访客写入
        String targetTopic = "dwd_traffic_unique_visitor_detail";
        SingleOutputStreamOperator<String> stringDs = filterDs2.map(jsonObj -> jsonObj.toJSONString());
        stringDs.print();
        stringDs.sinkTo(FlinkSinkUtil.getKafkaSink(targetTopic));
    }
}
